package week5.work2

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object distcp {

  type OptionMap = Map[Symbol,Any]
  //判断命令语句并分析
  def nextOption(map : OptionMap, list: List[String]):OptionMap = {
    def isSwitch(s : String) = (s(0) == '-')
    list match {
      case Nil => map // Nil是空List 表示无元素的情况
      case "-i" :: value => nextOption(map ++ Map('ignoreFailure -> 1), list.tail)
      case "-m" :: value :: tail => nextOption(map ++ Map('maxconcurrency -> value.toInt),tail)
      case string :: Nil  => nextOption(map ++ Map('outfile -> string), list.tail)
      case string :: tail => nextOption(map ++ Map('infile -> string), tail)
      case option :: opt2 :: tail if isSwitch(opt2) =>
        println("Unknown option "+option)
        sys.exit(1)
    }
  }
//便利所有文件路径
  def traverseDir(hdconf: Configuration, path: String, filePaths: ListBuffer[String]): Unit ={
    val files = FileSystem.get(hdconf).listStatus(new Path(path))
    files.foreach{
      fStatus =>{
        if(!fStatus.isDirectory){
          filePaths += fStatus.getPath.toString
        }else if(fStatus.isDirectory){
          traverseDir(hdconf, fStatus.getPath.toString, filePaths)
        }
      }
    }
  }

  def main(args: Array[String]): Unit = {
    if(args.length == 0){
      printf("No parameter!")
      return
    }
    val argList = args.toList
    val options = nextOption(Map(),argList)
    println(options)

    val sourceFolder = String.valueOf(options(Symbol("infile")))
    val targetFolder = String.valueOf(options(Symbol("outfile")))
    val concurrency =  (options(Symbol("maxconcurrency"))).toString.toInt
    val ignoreFailure = options(Symbol("ignoreFailure")).toString.toInt

    val sparkConf = new SparkConf().setAppName("distscp").setMaster("local[1]")
    val sc = new SparkContext(sparkConf)
    val sb = new StringBuffer();
    var fileNames = new ListBuffer[String]()

    val conf = new Configuration();
    traverseDir(conf, sourceFolder, fileNames)
    fileNames.foreach(
      fileName =>
        try {
          println("fileName replace is:" + fileName.replaceAll(sourceFolder,targetFolder))
          sc.textFile(fileName, concurrency).saveAsTextFile(fileName.replaceAll(sourceFolder, targetFolder))
        }catch {
          case t: Throwable => t.printStackTrace()
            if(ignoreFailure==0) {
              throw new Exception("failed to copy " + fileName)
            }
        }
    )
  }
}
